package sink

import java.util

import bean.SensorReading
import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.connectors.elasticsearch.{ElasticsearchSinkFunction, RequestIndexer}
import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink
import org.apache.flink.streaming.connectors.redis.common.mapper.{RedisCommand, RedisCommandDescription, RedisMapper}
import org.apache.http.HttpHost
import org.elasticsearch.action.index.IndexRequest
import org.elasticsearch.client.Requests

/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  * @date:2021/11/13 23:32
  *
  *
  *
  */
object EsSink {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    //file source
    val inputPath: String = "src/main/resources/sensor.txt"
    val fileStream: DataStream[String] = env.readTextFile(inputPath)

    //先转换为特定的类型
    val dataStream: DataStream[SensorReading] = fileStream.map(data => {
      val arr = data.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })
    dataStream.print()

    //定义HttpHosts
    val httpHosts = new util.ArrayList[HttpHost]();
    httpHosts.add(new HttpHost("192.168.88.106", 9200))

    dataStream.addSink(new ElasticsearchSink
    .Builder[SensorReading](httpHosts, new MyEsSinkFunction)
      .build())

    env.execute("es sink");


  }

  //定义写入一个es的EsSinkFunction
  class MyEsSinkFunction extends ElasticsearchSinkFunction[SensorReading] {
    override def process(t: SensorReading, runtimeContext: RuntimeContext, requestIndexer: RequestIndexer): Unit = {
      //包装一个map作为data source
      val dataSource = new util.HashMap[String, String]()
      dataSource.put("id", t.id)
      dataSource.put("temperature", t.temperature.toString)
      dataSource.put("timestamp", t.timestamp.toString)

      //创建index request，用户发送http请求
      val indexRequest: IndexRequest = Requests.indexRequest()
        .index("sensor-scala")
        .`type`("rendingdata")
        .source(dataSource)
      indexRequest

      //用indexer发送请求
      requestIndexer.add(indexRequest)
    }
  }

}
